package com.atguigu.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.utils.DateFormatUtil;
import com.atguigu.utils.KafkaUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.joda.time.DateTimeUtils;

import java.text.SimpleDateFormat;

//数据流:web/app -> logfile -> flume -> kafka(ODS) -> FlinkApp -> Kafka(DWD)
//程 序：Mock -> logfile -> f1.sh -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK)
/*
流量域事务事实表:
	流量域页面浏览事务事实表
	流量域启动事务事实表
	流量域动作事务事实表
	流量域曝光事务事实表
	流量域错误事务事实表

 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //todo 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //        // 需要从Checkpoint或者Savepoint启动程序 需启动hdfs
//        //开启Checkpoint,每隔5秒钟做一次CK  ,并指定CK的一致性语义
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        //设置超时时间为 1 分钟
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        //设置两次重启的最小时间间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        //设置任务关闭的时候保留最后一次 CK 数据
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        //指定从 CK 自动重启策略
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(1L), Time.minutes(1L)));
//        //设置状态后端
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/flinkCDC/220828");
//        //设置访问HDFS的用户名
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //todo 2.读取kafka topic_log主题数据创建流
        String topic_log = "topic_log";
        String groupId = "base_log_app_220828";
        DataStreamSource<String> kafkaLogDS = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topic_log, groupId));

        /*
        value:日志数据，经过上述操作变成了String类型
        {
    "actions":[
        {
            "action_id":"favor_add",
            "item":"33",
            "item_type":"sku_id",
            "ts":1651303989298
        },
        {
            "action_id":"cart_add",
            "item":"33",
            "item_type":"sku_id",
            "ts":1651303993596
        }
    ],
    "common":{
        "ar":"440000",
        "ba":"iPhone",
        "ch":"Appstore",
        "is_new":"0",
        "md":"iPhone X",
        "mid":"mid_3442864",
        "os":"iOS 12.4.1",
        "uid":"928",
        "vc":"v2.1.134"
    },
    "displays":[
        {
            "display_type":"promotion",
            "item":"19",
            "item_type":"sku_id",
            "order":1,
            "pos_id":3
        },
        {
            "display_type":"query",
            "item":"27",
            "item_type":"sku_id",
            "order":2,
            "pos_id":4
        }
    ],
    "page":{
        "during_time":17195,
        "item":"33",
        "item_type":"sku_id",
        "last_page_id":"good_list",
        "page_id":"good_detail",
        "source_type":"promotion"
    },
    "ts":1651303985000
}
         */
        //todo 3.过滤出非空以及合法json字符串数据并转为json对象（数据传输过程中可能会出现部分数据丢失的情况，导致 JSON 数据结构不再完整，因此需要对脏数据进行过滤。）
        SingleOutputStreamOperator<JSONObject> jsonObjLogDS = kafkaLogDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    try {
                        JSONObject jsonObjectValue = JSON.parseObject(value);
                        out.collect(jsonObjectValue);
                    } catch (JSONException e) {
                        System.out.println("脏数据（非法json）:" + value);
                    }

                }

            }
        }).name("").uid("");

        //todo 4.按照mid分组
        //注意！！！ keyedStream<JSONObject, String> -><value,key>
        KeyedStream<JSONObject, String> keyedStream = jsonObjLogDS.keyBy(jsonObjLog -> jsonObjLog.getJSONObject("common").getString("mid"));


        //todo 5.使用状态编程的方式实现新老用户校验
        //用富函数是因为获取状态需要用到富函数里的open方法里的getRuntimeContext()调出状态
        SingleOutputStreamOperator<JSONObject> jsonObjectWithNewFlagDS = keyedStream.map(new RichMapFunction<JSONObject, JSONObject>() {

            //定义状态，里面放日期
            private ValueState<String> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                //注意ValueStateDescriptor<String>里面的泛型必须手动添加
                valueState = getRuntimeContext().getState(new ValueStateDescriptor<String>("last-dt", String.class));
            }

            @Override
            public JSONObject map(JSONObject value) throws Exception {

                //获取状态日期
                String lastDtState = valueState.value();//上一次登录时间（准确来说是之前登录之间，不一定是上一次的）

                //获取当前数据中的is_new标记和日期
                String isNew = value.getJSONObject("common").getString("is_new");
                //注意：不要用下面这种获取日期的方式，多线程下会有错误，本案例是mapfunction，要用也可以，数据是一条一条的执行的
//                SimpleDateFormat simpleDateFormat = new SimpleDateFormat();
//                simpleDateFormat.parse(xxx);
//                simpleDateFormat.format(xxx)
                //自定义工具类
                String curDt = DateFormatUtil.toDate(value.getLong("ts"));


                //新老用户校验
                if ("1".equals(isNew)) {//前端发来1不一定是新用户
                    if (lastDtState == null || lastDtState.equals(curDt)) {
                        //lastDtState == null说明是该设备的第一条数据，lastDtState.equals(curDt)说明是今天，并且第一次登录在今天，今天多次登录都表示是新用户，都表示是新用户
                        valueState.update(curDt);
                    } else {//lastDtState != null && 登录日期不等于状态日期，都表示是老用户，要把value里的is_new改为0再返回
                        value.getJSONObject("common").put("is_new", "0");
                        //状态日期就不用更新了（如果更新为今天，今天卸载，又重新登录，则前端发来1，登录日期=状态日期，走上面的方法会被认为是新用户，是错误的）
                    }

                } else {//isNEW=0 前端发来0就一定是老用户
                    if (lastDtState == null) {//老用户并且状态为null，说明实时上线之前就登录了
                        valueState.update("1970-01-01");

                    }

                }


                return value;//返回的value的is_new是正确的is_new


            }
        });

        //todo 6.使用侧输出流的方式分流5个流  将页面日志从主流输出（来的数据既有页面日志也有启动日志），其他的放入侧输出流输出
        //注意！！！  定义侧输出流要有{}
        OutputTag<String> startTag = new OutputTag<String>("start") {

        };//启动//String侧输出流的类型，即要写到kafka的类型，也可以写JsonObject
        OutputTag<String> errorTag = new OutputTag<String>("error") {

        };//错误
        OutputTag<String> displayTag = new OutputTag<String>("display") {

        };//曝光
        OutputTag<String> actionTag = new OutputTag<String>("action") {

        };//动作

        //侧输出流只能用process
        SingleOutputStreamOperator<String> pageDS = jsonObjectWithNewFlagDS.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {

                //处理error错误日志
                String err = value.getString("err");
                if (err != null) {//*********************************
                    //说明value包含err字段，有err信息
                    ctx.output(errorTag, value.toJSONString());//通过侧输出流errorTag将value以json字符串的方式发送出去

                    //将错误信息移除，然后再过滤出start和放到各自的侧输出流里
                    value.remove("err");
                }

                //尝试获取启动信息（此时的信息要么是没有error字段，要么有，但前面删除了）
                String start = value.getString("start");
                if (start != null) {//启动日志
                    ctx.output(startTag, value.toJSONString());//获取包含start字段且不为start对应的value不为null的整条日志，包含common、ts字段，不包含错误信息（错误信息不存在或被移除了）
                } else {//页面日志（不含start字段）

                    //获取页面id和曝光时间
                    String pageId = value.getJSONObject("page").getString("page_id");
                    Long ts = value.getLong("ts");

                    //尝试获取曝光数据
                    JSONArray displays = value.getJSONArray("displays");
                    if (displays != null && displays.size() > 0) {//后者是为了不要displays：[],另外displays.size()不会出现空指针异常，因为前面判断成功了，后面就不会判断了
                        //遍历输出
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject displayJSONObject = displays.getJSONObject(i);//下标从0开始
                            //上面获取JSONObject的原因是还要额外放字段，JSONObject有put方法
                            displayJSONObject.put("page_id", pageId);
                            displayJSONObject.put("ts", ts);
                            //将数据放进侧输出流里
                            ctx.output(displayTag, displayJSONObject.toJSONString());//获取仅仅display信息，并且添加了页面id和ts


                        }

                        //此时的页面日志没有err信息（没有或移除了），但是还包含displays信息，不删没关系，因为后面只取actions字段就可以了
                        //尝试获取动作数据
                        JSONArray actions = value.getJSONArray("actions");
                        if (actions != null && actions.size() > 0) {//后者是为了不要actions：[],另外actions.size()不会出现空指针异常，因为前面判断成功了，后面就不会判断了
                            //遍历输出
                            for (int i = 0; i < actions.size(); i++) {
                                JSONObject actionJSONObject = actions.getJSONObject(i);//下标从0开始
                                //上面获取JSONObject的原因是还要额外放字段，JSONObject有put方法
                                actionJSONObject.put("page_id", pageId);
                                ctx.output(actionTag, actionJSONObject.toJSONString());//获取仅仅动作信息，并且添加了page_id，动作信息自带动作时间，所以不用添加


                            }


                        }
                    }


                    //过滤出页面日志，没有错误信息（没有或移除了），有common、ts字段，然后过滤掉displays和actions
                    value.remove("displays");
                    value.remove("actions");
                    out.collect(value.toJSONString());

                }
            }
        });



        //todo 7.获取侧输出流的数据并写到kafka
        DataStream<String> startDS = pageDS.getSideOutput(startTag);
        DataStream<String> errorDS = pageDS.getSideOutput(errorTag);
        DataStream<String> displayDS = pageDS.getSideOutput(displayTag);
        DataStream<String> actionDS = pageDS.getSideOutput(actionTag);
//        DataStream<String> actions = pageDS.getSideOutput(new OutputTag<String>("actions"));

        //打印看一看
        startDS.print("startDS>>>>");
        errorDS.print("errorDS>>>>");
        displayDS.print("displayDS>>>>");
        actionDS.print("actionDS>>>>");
        pageDS.print("pageDS>>>>");

//        startDS.addSink(new FlinkKafkaProducer<String>(XXX))//写一个工具类
        //定义各自写入kafka的主题
        String start_topic="dwd_traffic_start_log";
        String error_topic="dwd_traffic_error_log";
        String display_topic="dwd_traffic_display_log";
        String action_topic="dwd_traffic_action_log";
        String page_topic="dwd_traffic_page_log";

        //事实表的流量域的5个分类的日志写到kafka里
        startDS.addSink(KafkaUtil.getFlinkKafkaProducer(start_topic));
        errorDS.addSink(KafkaUtil.getFlinkKafkaProducer(error_topic));
        displayDS.addSink(KafkaUtil.getFlinkKafkaProducer(display_topic));
        actionDS.addSink(KafkaUtil.getFlinkKafkaProducer(action_topic));
        pageDS.addSink(KafkaUtil.getFlinkKafkaProducer(page_topic));



        //todo 8.启动任务
        env.execute("baselogapp");//FLINK任务的名字，不加的话所有任务都叫flinkstreamingjob

    }
}

